import 'dart:async';
import 'dart:ffi';
import 'dart:typed_data';
import 'dart:ui';
import 'package:flutter/cupertino.dart';
import 'package:flutter_mqtt/src/utils.dart';
import 'package:mqtt_client/mqtt_client.dart';
import 'package:mqtt_client/mqtt_server_client.dart';

import 'mqtt_config.dart';
import 'mqtt_message.dart';

class MqttHelper {
  static const String tag = 'MqttHelper';

  late MqttConfig config;

  // MQTT客户端
  late MqttServerClient _client;

  bool _isConnected = false;

  //是否正在重新连接中
  bool _isAutoConnectIng = false;

  bool get isAutoConnectIng => _isAutoConnectIng;

  // 连接状态
  bool get isConnected => _isConnected && _client.connectionStatus?.state == MqttConnectionState.connected;

  // 消息流控制器
  late StreamController<MqttMessageData> _messageStreamController;

  Stream<MqttMessageData> get messageStream => _messageStreamController.stream;

  //连接流控制器
  late StreamController<bool> _connectionStreamController;

  Stream<bool> get connectionStream => _connectionStreamController.stream;

  //连接错误控制器
  late StreamController<dynamic> _errorStreamController;

  Stream<dynamic> get errorStream => _errorStreamController.stream;

  //订阅控制器
  late StreamController<MapEntry<String, bool>> _subscriptionStreamController;

  Stream<MapEntry<String, bool>> get subscriptionStream => _subscriptionStreamController.stream;

  // 主题过滤器集合
  final Set<String> _topicFilters = {};
  bool _isInit = false;

  MqttHelper(this.config);

  void _assertInit() {
    assert(_isInit, 'MqttHelper must be initialized before use');
  }

  void _assertConnected() {
    assert(isConnected, 'MqttHelper must be connected before use');
  }

  // 初始化MQTT客户端
  void initialize() {
    if (_isInit) return;
    _isInit = true;
    //初始化流
    _connectionStreamController = StreamController.broadcast();
    _messageStreamController = StreamController.broadcast();
    _errorStreamController = StreamController.broadcast();
    _subscriptionStreamController = StreamController.broadcast();

    _client = MqttServerClient.withPort(config.server, config.clientId, config.port)
      ..logging(on: false)
      ..autoReconnect = config.enableAutoReconnect
      ..keepAlivePeriod = config.heartbeatInterval
      ..onDisconnected = _onDisconnectedCallback
      ..onAutoReconnect = () {
        _isAutoConnectIng = true;
        // 自动连接中，重复回调
        print("$tag : auto reconnect status ${_client.connectionStatus}");
      }
      ..onAutoReconnected = () {
        _isAutoConnectIng = false;
        //自动连接成功，在onConnected之后
        print("$tag : auto reconnected");
      }
      ..onConnected = () {
        print("$tag : connected");
      }
      ..onSubscribed = (topic) {
        print("$tag : subscribed topic $topic");
        _subscriptionStreamController.add(MapEntry(topic, true));
      }
      ..onUnsubscribed = (topic) {
        print("$tag : unsubscribed topic $topic");
        _subscriptionStreamController.add(MapEntry(topic ?? '', false));
      }
      ..onSubscribeFail = (topic) {
        print("$tag : subscribe fail topic $topic");
        _subscriptionStreamController.add(MapEntry(topic, false));
      }
      ..onFailedConnectionAttempt = (status) {
        print("$tag : failed connection attempt $status");
      }
      ..setProtocolV311();

    // 认证配置
    final connMess = MqttConnectMessage().withWillQos(config.defaultQos).startClean();

    if (config.username != null) {
      connMess.authenticateAs(config.username!, config.password ?? '');
    }

    _client.connectionMessage = connMess;
  }

  // 连接MQTT服务器
  Future<bool> connect() async {
    try {
      final status = await _client.connect();
      if (status?.state == MqttConnectionState.connected) {
        _isConnected = true;
        _connectionStreamController.add(true);
        _setupMessageListener();
        return true;
      } else {
        print("$tag : $status");
        _connectionStreamController.add(false);
        _errorStreamController.add(status);
        return false;
      }
    } catch (e) {
      print("$tag : $e");
      _connectionStreamController.add(false);
      _errorStreamController.add(e);
      return false;
    }
  }

  // 消息监听设置
  void _setupMessageListener() {
    _client.updates!.listen((messages) {
      for (var msg in messages) {
        final payload = msg.payload as MqttPublishMessage;
        final mqttMessage = MqttMessageData(topic: msg.topic, message: payload.payload.message.toList());
        _handleFilteredMessage(mqttMessage);
      }
    });
  }

  // 主题过滤器处理
  void _handleFilteredMessage(MqttMessageData message) {
    for (var filter in _topicFilters) {
      if (topicMatchesFilter(message.topic, filter)) {
        _messageStreamController.add(message);
        break;
      }
    }
  }

  // 断开连接回调
  void _onDisconnectedCallback() {
    _client.disconnect();
    _isConnected = false;
    _isAutoConnectIng = false;
    _topicFilters.clear();
    _connectionStreamController.add(false);
  }

  // 断开连接
  void disconnect() {
    _assertInit();
    _isConnected = false;
    _isAutoConnectIng = false;
    _client.disconnect();
  }

  // 关闭，释放资源，下次必须调用initialize
  void close() {
    _isInit = false;
    disconnect();
    //清除所有流控制器
    _connectionStreamController.close();
    _messageStreamController.close();
    _errorStreamController.close();
    _subscriptionStreamController.close();
  }

  // 订阅（支持通配符）
  void subscribe(String topicFilter, {MqttQos? qos}) {
    _assertInit();
    _assertConnected();
    _client.subscribe(topicFilter, qos ?? config.defaultQos);
    _topicFilters.add(topicFilter);
  }

  // 取消订阅
  void unsubscribe(String topicFilter) {
    _assertInit();
    _assertConnected();
    _client.unsubscribe(topicFilter);
    _topicFilters.remove(topicFilter);
  }

  // 发布
  void publish(
    String topic,
    String message, {
    bool retain = false,
    MqttQos? qos,
  }) {
    _assertInit();
    _assertConnected();
    final builder = MqttClientPayloadBuilder();
    builder.addString(message);
    _client.publishMessage(
      topic,
      qos ?? config.defaultQos,
      builder.payload!,
      retain: retain,
    );
  }
}
